package com.flink.sync.oracle;


import org.apache.commons.dbcp2.BasicDataSource;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.PreparedStatement;

public class Sink2Oracle extends RichSinkFunction<TQrUser> {
	private PreparedStatement ps;
	private BasicDataSource dataSource;
	private Connection connection;

	/**
	 * open() 方法中建立连接，这样不用每次 invoke 的时候都要建立连接和释放连接
	 * @param parameters
	 * @throws Exception
	 */
	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		dataSource = new BasicDataSource();
		connection = getConnection(dataSource);
		String sql = "insert into osoadata.t_qr_user(id, idno, mobileno, qrcode, status, modify_date, name, yxwd, create_date) values(?, ?, ?, ?, ?, ?, ?, ?, ?);";
		ps = this.connection.prepareStatement(sql);
	}

	private Connection getConnection(BasicDataSource dataSource) {
		dataSource.setDriverClassName("oracle.jdbc.OracleDriver");
		dataSource.setUrl("jdbc:oracle:thin:@10.1.171.84:1521:ORCL");
		dataSource.setUsername("osoadata");
		dataSource.setPassword("888888");
		//设置连接池的一些参数
		dataSource.setInitialSize(10);
		dataSource.setMaxTotal(100);
		dataSource.setMinIdle(10);

		Connection conn = null;
		try {
			conn = dataSource.getConnection();
		} catch (Exception e) {
			System.out.println("-----------oracle get connection has exception , msg = " + e.getMessage());
		}
		return conn;
	}

	@Override
	public void close() throws Exception {
		super.close();
		//关闭连接和释放资源
		if (connection != null) {
			connection.close();
		}
		if (ps != null) {
			ps.close();
		}
	}

	/**
	 * 每条数据的插入都要调用一次 invoke() 方法
	 *
	 * @param value
	 * @param context
	 * @throws Exception
	 */
	@Override
	public void invoke(TQrUser value, Context context) throws Exception {
		ps.setLong(1, value.getId());
		ps.setString(2, value.getIdno());
		ps.setString(3, value.getMobileno());
		ps.setString(4, value.getQrcode());
		ps.setString(5, value.getStatus());
		ps.setString(6, value.getModify_date());
		ps.setString(7, value.getName());
		ps.setString(8, value.getYxwd());
		ps.setString(9, value.getCreate_date());

		int cnt = ps.executeUpdate();

		System.out.println("成功插入行数：" + cnt);
	}
}
